package com.darcytech.debezium.core.api.impl;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import com.darcytech.debezium.common.utils.ObjectMapperUtils;
import com.darcytech.debezium.common.utils.StringLockUtils;
import com.darcytech.debezium.core.rpc.proto.Rpc;

import io.netty.channel.socket.SocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class ApiSender {

    private static final Logger LOGGER = LoggerFactory.getLogger(ApiSender.class);

    private final static int MAX_RETRY_TIMES = 2;

    private final ConcurrentMap<String, String> resMap = new ConcurrentHashMap<>();

    protected volatile SocketChannel channel;

    public void setChannel(SocketChannel channel) {
        this.channel = channel;
    }

    public void setRes(String requestId, String data) {
        resMap.put(requestId, data);
    }

    protected <T> T send(Object data, String type, Class<T> returnClazz) throws IOException, InterruptedException {
        for (int i = 0; ; i++) {
            try {
                return send0(data, type, returnClazz);
            } catch (Throwable cause) {
                if (i < MAX_RETRY_TIMES) {
                    if (cause instanceof IOException || cause instanceof InterruptedException) {
                        LOGGER.error("request type:{} happen IOException or InterruptedException", type);
                    } else {
                        LOGGER.error("request type:{} happen:", type, cause);
                        throw cause;
                    }
                } else {
                    LOGGER.error("request type:{} happen:", type, cause);
                    throw cause;
                }
            }
        }

    }


    private <T> T send0(Object data, String type, Class<T> returnClazz) throws IOException, InterruptedException {
        String requestId = UUID.randomUUID().toString();
        String jsonData = ObjectMapperUtils.writeAsString(data);
        Rpc.Req req = Rpc.Req.newBuilder().setId(requestId).setType(type).setData(jsonData).build();

        String respJson;
        try {
            Object waitDataLock = StringLockUtils.getOrCreateStringLock(requestId);
            this.channel.writeAndFlush(req);
            synchronized (waitDataLock) {
                respJson = resMap.get(requestId);
                if (respJson == null) {
                    try {
                        waitDataLock.wait(5000);
                    } catch (InterruptedException e) {
                        respJson = resMap.get(requestId);
                        if (respJson == null) {
                            throw new InterruptedException("Interrupted");
                        }
                    }
                    respJson = resMap.get(requestId);
                    if (respJson == null) {
                        throw new IOException("network anomaly");
                    }
                }
            }
        } finally {
            StringLockUtils.removeStringLock(requestId);
        }

        return ObjectMapperUtils.readValue(respJson, returnClazz);
    }

}
